package net

/*
  是一个数据生产者
*/

import (
	"dgo/MyProj/PrjConf"
	"dgo/MyProj/PrjHttpUtils"
	"dgo/goutils/dnet"
	"dgo/goutils/gis"
	"dgo/goutils/utils"
	"fmt"
	"gitee.com/ymofen/supervalue"
	"sync/atomic"
	"time"
)

type ProjTcpClientPlug struct {
	PlgBase
	tcpclt         *dnet.TcpClt
	last_dis       string
	nmea           string
	nmea_task_flag int32
	releaseflag    int
	conn_n         int

	outputEx *PlgBaseInnerEx
}

func NewProjTcpClientPlug() *ProjTcpClientPlug {
	rval := &ProjTcpClientPlug{}
	return rval
}

/*
**

	响应HttpStatus请求
*/
func (this *ProjTcpClientPlug) OnHttpStatus(rec *PlgHttpRec) (rval interface{}, errcode int, err error) {
	json := wrapper.NewSVObject()
	json.SetKeyValues("plugid", this.PlugId, "uuid", PrjConf.APP_UUID,
		"remote-addr", this.tcpclt.GetRemoteAddr(),
		"recv-monitor", this.tcpclt.RecvMonitor.Info(),
		"conn-n", this.conn_n,
		"lastdis", this.last_dis)
	PrjHttpUtils.JSON2TableRec(rec.RespW, "", json.AsJSONString(false))
	return
}

/*
*

	{
	  "input":{        // 如果指定了input则将订阅到的数据通过tcpclt进行发送
	     "id":"",
	     "topic":"topic"
	   },
	  "remoteaddr":"127.0.0.1:2101",
	  "nmea":"$GNGGA,010918.00,3025.36715396,N,11427.95302424,E,2,12,0.9,28.743,M,-13.869,M,85.0,0129*71",
	  "output":{

	  }
	}
*/
func (this *ProjTcpClientPlug) ReloadConf(conf *wrapper.SuperValue) (err error) {
	if this.outputEx == nil {
		this.outputEx = PlgNewInnerEx()
	}
	if this.tcpclt == nil {
		this.tcpclt = dnet.NewTcpClt("")
		this.tcpclt.AfterRecvBuf = this.OnRecvNtripCltBuffer
		this.tcpclt.RecvMonitor = utils.NewMonitorSpeedSizeRec()
		utils.Monitor.AddMonitor(this.tcpclt.RecvMonitor)
		this.tcpclt.AfterDisconnected = func(sender *dnet.TcpClt) {
			this.last_dis = fmt.Sprintf("[%s]%s", utils.NowString(), sender.DisReason)
		}
		this.tcpclt.AfterConnected = func(sender *dnet.TcpClt) {
			this.conn_n++
		}
	}
	changed := false
	conf.CheckStrFunc("remoteaddr", func(strv string) {
		utils.ReplacePlaceholder(strv, "$$", "$$")
		if this.tcpclt.SetAddress(strv) {
			changed = true
		}
	})

	conf.CheckStrFunc("nmea", func(strv string) {
		this.nmea = strv
		if len(this.nmea) > 0 {
			if atomic.CompareAndSwapInt32(&this.nmea_task_flag, 0, 1) {
				utils.AddTaskInterval(this.PlugId+"_nmea_task", func() bool {
					if len(this.nmea) > 0 && this.tcpclt.Active() {
						nmea := gis.NewNMEA()
						r := nmea.ParseText(this.nmea)
						if r == -1 {
							this.tcpclt.Write([]byte(this.nmea))
						} else {
							nmea.UpdateUTCTime()
							str := nmea.Encode() + "\r\n"
							this.tcpclt.Write([]byte(str))
						}
					}
					return this.releaseflag == 0
				}, time.Second*10, utils.ZeroTime)
			}
		}
	})

	if changed {
		if !this.tcpclt.Active() {
			this.tcpclt.Start()
		} else {
			this.tcpclt.RequestDisconnect("配置改变")
		}
	} else {
		err = utils.EUnChangedErr
	}

	this.ReloadStatusOConf(conf, func() {
		this.SendStatusFunc(func(json *wrapper.SuperValue) {
			json.SetKeyValues("plugid", this.PlugId, "uuid", PrjConf.APP_UUID,
				"build-time", PlgFnBuildTime(), "runsecs", PrjHttpUtils.GetRunSecs())
		})
	}, func() {
		//utils.Monitor.AddMonitorRec(fmt.Sprintf("plugin_%d_recv", this.PlugDNA), &this.recvMonitor)
	}, func() {
		//utils.Monitor.RemoveMonitorRec(fmt.Sprintf("plugin_%d_recv", this.PlugDNA))
	})

	this.outputEx.ReloadConf(conf.ValueByName("output"))

	this.ReloadInputConf(conf, this.OnInputSubscribeCallback)

	return
}

func (this *ProjTcpClientPlug) OnRecvNtripCltBuffer(sender *dnet.TcpClt, buffer []byte) {
	this.outputEx.OutputTaskBuffer(buffer)
}

func (this *ProjTcpClientPlug) OnInputSubscribeCallback(sender interface{}, channelid string, subscribeid interface{}, data interface{}, tag interface{}, isBreak *bool) bool {
	tcpclt := this.tcpclt
	if tcpclt != nil {
		if buf, ok := data.([]byte); ok {
			tcpclt.Write(buf)
		}
	}
	return true
}

func (this *ProjTcpClientPlug) OnObjectNotify(sender interface{}, notifytype int, data interface{}) (rval interface{}, err error) {
	if notifytype == utils.NOTIFY_TYPE_SETCONF {
		if conf, ok := data.(*wrapper.SuperValue); ok {
			err = this.ReloadConf(conf)
		} else {
			err = fmt.Errorf("非法的JSON配置")
		}
	} else if notifytype == utils.NOTIFY_TYPE_RELEASE {
		this.releaseflag = 1
		this.outputEx.Release()
		if this.tcpclt != nil {
			utils.Monitor.RemoveMonitorRec(this.tcpclt.RecvMonitor)
		}
	}
	return
}

func CreateProjTcpClientPlugFunc(args ...interface{}) (rval interface{}, err error) {
	obj := NewProjTcpClientPlug()
	if len(args) > 0 {
		if str, ok := args[0].(string); ok {
			obj.PlugId = str
		}
	}
	rval = obj
	return
}
